package com.soft.small.util.mqtt;

import com.soft.small.util.encrypt.SignUtil;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.internal.wire.MqttWireMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;


/**
 * mqtt工具类
 */
@Data
@Slf4j
public class MyMqttClient {
    private String username;
    private String password;
    private String host;
    private String clientId;
    private MqttClient mqttClient;
    private String appKey;
    private String appSecret;


    public MyMqttClient(String host, String clientId, String username, String password){
        this.host = host;
        this.clientId = clientId;
        this.username = username;
        this.password = password;
    }

    /**
     * mqttClient connect mqtt server
     *
     */
    public void setMqttClient(MqttConnectOptions options, MqttCallback mqttCallback) {
        try {
            mqttClient = new MqttClient(host,clientId,new MemoryPersistence());
            mqttClient.setCallback(mqttCallback);
            mqttClient.connect(options);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    public void close(String type, String data) {
        publish(type,data);
        try {
            if (mqttClient != null){
                mqttClient.disconnect();
                mqttClient.close();
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * mqttClient publish message default qos 1
     *
     * @param topic
     * @param msg
     * @throws MqttException
     * @return
     */
    public MqttWireMessage publish(String topic, String msg) {
        log.info("客户端发布到topic:{}的内容是：{}",topic,msg);
        MqttMessage mqttMessage = new MqttMessage();
        byte[] send = SignUtil.getPublishMessage(msg,appSecret);
        mqttMessage.setPayload(send);
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token;
        try {
            token = mqttTopic.publish(mqttMessage);
            token.waitForCompletion();
            log.info("消息发布状态：{}",token.isComplete());
            return token.getResponse();
        } catch (MqttException e) {
            e.printStackTrace();
        }

        return null;
    }

    /**
     * mqttClient publish message with qos
     *
     * @param topic
     * @param msg
     * @param qos   0,1,2
     * @throws MqttException
     */
    public void publish(String topic, String msg, int qos) throws MqttException {
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(qos);
        mqttMessage.setPayload(msg.getBytes());
        MqttTopic mqttTopic = mqttClient.getTopic(topic);
        MqttDeliveryToken token = mqttTopic.publish(mqttMessage);
        token.waitForCompletion();
    }

    /**
     * mqttClient subscribe topic default qos 1
     *
     * @param topic
     * @throws MqttException
     */
    public void subscribe(String topic){
        try {
            mqttClient.subscribe(topic);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }

    /**
     * mqttClient subscribe topic with qos
     *
     * @param topic
     * @param qos
     * @throws MqttException
     */
    public void subscribe(String topic, int qos) throws MqttException {
        mqttClient.subscribe(topic, qos);
    }

}
